package tech.spiro.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.nio.CharBuffer;
import java.util.Map;
import java.util.Set;

public class PerfLogMessageFlatten implements Serializable {

    Logger logger = LoggerFactory.getLogger(PerfLogMessageFlatten.class);

    public boolean messageFieldDot2Underline = true;
    private static final String MESSAGE_FIELD_NAME_IN_RECORD = "message";
    private static final String RAW_RECORD_FIELD_NAME = "raw_record";

    public void setMessageFieldDot2Underline(boolean messageFieldDot2Underline) {
        this.messageFieldDot2Underline = messageFieldDot2Underline;
    }

    public byte[] flattenMessage(byte[] value) {

        String record = new String(value);

        JSONObject resultJson = new JSONObject();
        JSONObject jsonObject;
        Set<Map.Entry<String, Object>> entrySet = null;
        try {
            jsonObject = JSON.parseObject(record);
            entrySet = jsonObject.entrySet();

            for (Map.Entry<String, Object> entry : entrySet) {

                if (!entry.getKey().equalsIgnoreCase(MESSAGE_FIELD_NAME_IN_RECORD)) {
                    resultJson.put(entry.getKey(), entry.getValue());
                    continue;
                }
                processMessage(entry.getValue(), resultJson);
            }

        } catch (Exception e) {
            logger.error("error when parse record.", e);
            resultJson = new JSONObject();
            resultJson.put(RAW_RECORD_FIELD_NAME, record);
        }


        return JSON.toJSONBytes(resultJson);
    }

    enum State {
        KV_FINDING,
        KEY_SCANNING,
        VALUE_SCANNING
    }

    public static final char TOKEN_LT = '<';
    public static final char TOKEN_GT = '>';
    public static final char TOKEN_EQ = '=';

    private State state = State.KV_FINDING;

    private String message = null;

    private CharBuffer keyBuf = null;
    private CharBuffer valueBuf = null;

    private int keyStartIdx = 0;
    private int keyEndIdx = 0;
    private int valueStartIdx = 0;
    private int valueEndIdx = 0;

    private boolean valueHasGT = false;

    private void initialize() {
        keyBuf = null;
        valueBuf = null;
        keyStartIdx = 0;
        keyEndIdx = 0;
        valueStartIdx = 0;
        valueEndIdx = 0;
        valueHasGT = false;

        state = State.KV_FINDING;
    }

    private void onKVCompleted(Map<String, Object> targetMap) {
        // keyStartIdx == 0 happen on case "... <=value> ..."
        if ((keyStartIdx == 0) || (keyEndIdx - keyStartIdx <= 0)) {
            return;
        }

        // build key
        keyBuf = CharBuffer.wrap(this.message, keyStartIdx, keyEndIdx);
        String key = keyBuf.toString().trim();
        if (messageFieldDot2Underline) {
            key = key.replace('.', '_');
        }

        // build value and put to json object

        // valueStartIdx == 0 happen on case "... <key> ..."
        if (valueStartIdx == 0) {
            targetMap.put(key, "");
            return;
        }
        valueBuf = CharBuffer.wrap(this.message, valueStartIdx, valueEndIdx);
        targetMap.put(key, valueBuf.toString().trim());
    }

    public void processMessage(Object value, Map<String, Object> targetMap) throws Exception {
        initialize();

        this.message = (String) value;
        int chIdx = 0;
        while (chIdx < message.length()) {
            char ch = message.charAt(chIdx);
            int currIdx = chIdx;
            chIdx++;

            switch (state) {
                case KV_FINDING:
                    if (ch == TOKEN_LT) {
                        state = State.KEY_SCANNING;
                    }
                    break;
                case KEY_SCANNING:
                    if (ch == TOKEN_LT) {
                        // case: "... <<key=value> ..." or "... <abc<key=value> ..."
                        throw new Exception("key can not contain '<'");
                    }

                    if (ch == TOKEN_GT) {
                        keyEndIdx = currIdx;

                        // KV Completed
                        onKVCompleted(targetMap);
                        initialize();
                        continue;
                    }

                    if (ch == TOKEN_EQ) {
                        keyEndIdx = currIdx;
                        state = State.VALUE_SCANNING;
                        continue;
                    }

                    if (keyStartIdx == 0) {
                        keyStartIdx = currIdx;
                    }
                    break;
                case VALUE_SCANNING:

                    if (ch == TOKEN_LT && valueHasGT) {
                        // KV Completed
                        onKVCompleted(targetMap);
                        initialize();

                        // Next K/V Found
                        state = State.KEY_SCANNING;
                        continue;
                    }

                    if (ch == TOKEN_GT) {
                        valueHasGT = true;
                        valueEndIdx = currIdx;
                    }

                    if (valueStartIdx == 0) {
                        valueStartIdx = currIdx;
                    }

                    break;
            }
        }

        // To the End
        if (state == State.VALUE_SCANNING) {
            if (!valueHasGT) {
                // happen on case:   "... <key=value"
                valueEndIdx = message.length();
            }
            onKVCompleted(targetMap);
        }
    }

}
